package mapred.joinmode;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class UserJoinMapper extends Mapper<Object, Text, Text, Text> {
    private Text outkey = new Text();
    private Text outvalue = new Text();

    protected void map(Object key, Text value, Context context)
            throws IOException, InterruptedException {
//        // Parse the input string into a nice map
        // MRDPUtils.transformXmlToMap:将xml结构的数据，转换成Key-Value格式
//        <user>
//        <id>1</id>
//        <name>张三</name>
//        <age>34</age>
//        <birthday>1990-1-1</birthday>
//        </user>
        // 经过MRDPUtils.transformXmlToMap后，转换成了：
//        key=id,value=1
//        key=name，value=张三
//        key=age,value=34

//        Map<String, String> parsed = MRDPUtils.transformXmlToMap(value.toString());
//        String userId = parsed.get("Id");
//        // The foreign join key is the user ID
//        outkey.set(userId);
//        // Flag this record for the reducer and then output
//        outvalue.set("A" + value.toString());
//        context.write(outkey, outvalue);

        // 对每一行输入按照【\t】进行拆分
        String[] fields = value.toString().split("\t");
        // 取数组的第0个元素，这个元素的含义是当前行的主键
        String userid = fields[0];
        // 主键作为当前Mapper的输出Key的值
        outkey.set(userid);
        // 对于value，在前面加入了一个固定的前缀：【A】
        // 数据特征如下：
        // A1	张三	34	1990-1-1
        outvalue.set("A" + value.toString());

        // 当前Mapper输出的key：Value是：
        // 1：A1	张三	34	1990-1-1
        // 2：A2	李四	23	2000-1-1
        // 3：A3	王五	12	2010-3-3
        // 4：A4	赵六	33  0000-0-0
        context.write(outkey, outvalue);
    }
}